package com.patsnap.data.npd.dw.etl.utils;

import com.amazonaws.services.s3.AmazonS3Client;
import com.patsnap.data.npd.dw.repository.DBOperator;
import com.patsnap.one.etl.tool.s3.S3Helper;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.session.SqlSessionFactory;

import java.util.Map;

@Slf4j
public class FlinkDBOperatorUtils {

    private static final String DATASOURCE_NAME_NPD_ODS = "npd_ods";
    private static final String DATASOURCE_NAME_NPD_DWD = "npd_dwd";
    private static final String DATASOURCE_NAME_NPD_ADS = "npd_ads";
    private static final String DATASOURCE_NAME_SA_PRODUCTION = "npd_sa";

    public static DBOperator initDBOperator(Map<String, SqlSessionFactory> sqlSessionFactoryMap) {
        if (sqlSessionFactoryMap == null) {
            return null;
        }

        SqlSessionFactory odsSqlSessionFactory = null;
        if (sqlSessionFactoryMap.containsKey(DATASOURCE_NAME_NPD_ODS)) {
            odsSqlSessionFactory = sqlSessionFactoryMap.get(DATASOURCE_NAME_NPD_ODS);
        }

        SqlSessionFactory dwdSqlSessionFactory = null;
        if (sqlSessionFactoryMap.containsKey(DATASOURCE_NAME_NPD_DWD)) {
            dwdSqlSessionFactory = sqlSessionFactoryMap.get(DATASOURCE_NAME_NPD_DWD);
        }

        SqlSessionFactory adsSqlSessionFactory = null;
        if (sqlSessionFactoryMap.containsKey(DATASOURCE_NAME_NPD_ADS)) {
            adsSqlSessionFactory = sqlSessionFactoryMap.get(DATASOURCE_NAME_NPD_ADS);
        }

        SqlSessionFactory saSqlSessionFactory = null;
        if (sqlSessionFactoryMap.containsKey(DATASOURCE_NAME_SA_PRODUCTION)) {
            saSqlSessionFactory = sqlSessionFactoryMap.get(DATASOURCE_NAME_SA_PRODUCTION);
        }

        return new DBOperator(odsSqlSessionFactory, dwdSqlSessionFactory, adsSqlSessionFactory, saSqlSessionFactory);
    }

    public static AmazonS3Client initS3Client(Map<String, String> configMap) {
        if (configMap.containsKey("s3_datasource_names")) {
            String s3SourceName = configMap.get("s3_datasource_names");
            if (configMap.containsKey(s3SourceName + "_s3_endpoint")
                    && configMap.containsKey(s3SourceName + "_s3_access_key")
                    && configMap.containsKey(s3SourceName + "_s3_secret_key")) {
                String s3Endpoint = configMap.get(s3SourceName + "_s3_endpoint");
                String s3AccessKey = configMap.get(s3SourceName + "_s3_access_key");
                String s3SecretKey = configMap.get(s3SourceName + "_s3_secret_key");

                return S3Helper.createS3Client(s3AccessKey, s3SecretKey, null, s3Endpoint);
            } else {
                log.warn("s3 source {} config missed", s3SourceName);
            }
        }
        return null;
    }
}
